package com.zhu.app.dwd;

import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 用户优惠券领用事物事实表  用户领用一条优惠券
 * 业务数据库的优惠券领用表会新增一条数据，因此操作类型为
 * insert 的数据即为优惠券领取数据。
 */
public class DWDToolCouponGetApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(1); //  kafka partition 4

        //check point
           /*
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
        */
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);
        Configuration configuration = streamTableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.exec.state.ttl", "5 s");  //状态存活时间

        //todo kafka topic_db
        streamTableEnvironment.executeSql(ZhuKafkaUtil.getTopicDB("dwd_tool_coupon_get"));

        //todo coupon_use
        Table result_table = streamTableEnvironment.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['coupon_id'] coupon_id, " +
                        "data['user_id'] user_id, " +
                        "date_format(data['get_time'],'yyyy-MM-dd') date_id, " +
                        "data['get_time'] get_time " +
                        "from topic_db where `table` = 'coupon_use' " +

                        "and `type` = 'insert' "
        );
        streamTableEnvironment.createTemporaryView("result_table", result_table);


        // TODO  建立 Kafka-Connector dwd_tool_coupon_get 表
        streamTableEnvironment.executeSql("create table dwd_tool_coupon_get (\n" +
                "id string,\n" +
                "coupon_id string,\n" +
                "user_id string,\n" +
                "date_id string,\n" +
                "get_time string\n" +
                ")" +
                ZhuKafkaUtil.getKafkaSinkDDL("dwd_tool_coupon_get"));
// TODO 6. 将数据写入 Kafka-Connector 表
        streamTableEnvironment.executeSql("" +
                "insert into dwd_tool_coupon_get select * from result_table").print();

    //todo execute
        streamExecutionEnvironment.execute();


    }

}
